"""
套接字控制台实用对象

date: 2025/7/10
author: SiHeng Tang
file: web_api.py
copyright(c) DFSA Eastern Software
此程序不能提供任何担保 WITHOUT WARRANTY OF ANY KIND
"""

import logging
import queue
import threading
import time
from copy import deepcopy
from json import JSONDecodeError

from helpers.rrtlib import UnixSocketRRT, IPv4SocketRRT, SerialRRT
from mj_errors import MJRuntimeError

logger = logging.getLogger(__name__)

MJ_RPC_COMMON_RESULT = {
    "jsonrpc": "2.0",
    "result": list(),
    "id": int()
}

MJ_RPC_ERROR_RESULT = {
    "jsonrpc": "2.0",
    "error": dict(),
    "id": int()
}

MJ_RPC_TOPIC_UPDATE_RESULT = {
    "jsonrpc": "2.0",
    "result": str(),
    "id": int()
}

MJ_RPC_ERROR_PARSE_ERROR_CODE = -32700
MJ_RPC_ERROR_INVALID_REQUESTS_CODE = -32600
MJ_RPC_ERROR_MTD_NOT_FOUND_CODE = -32601
MJ_RPC_ERROR_INVALID_PARAMS_CODE = -32602
MJ_RPC_ERROR_INTERNAL_ERROR_CODE = -32603


# noinspection PyBroadException
class LoggingTopicHandler(logging.Handler):
    """
    日志话题订阅，此类继承日志处理类，实现代码基本照搬流处理器，重写输出逻辑为保存每一行输出。
    """
    terminator = '\n'

    def __init__(self, level: int):
        """
        初始化话题，自动添加日志句柄并设置日志级别和前缀
        :param level: 日志级别
        """
        logging.Handler.__init__(self)
        self.setLevel(level)
        self.setFormatter(logging.Formatter('[%(levelname)s]%(asctime)s:%(module)s:%(message)s'))

        self._len_lim = 500
        self._line_cache: list[str] = []

        logging.getLogger('').addHandler(self)

    def emit(self, record):
        try:
            msg = self.format(record)
            if len(self._line_cache) > self._len_lim:
                self._line_cache = self._line_cache[1:]
            self._line_cache.append(msg)

        except RecursionError:
            raise

        except Exception:
            self.handleError(record)

    def topic_SYS_LOG(self, rpc_params: str | list):
        """
        日志话题
        :param rpc_params: 话题参数
        :return: 返回已经缓存的日志行
        """
        if rpc_params == "subscribe":
            return 0

        if rpc_params == "unsubscribe":
            return 0

        if rpc_params == "mailbox_update":
            res = deepcopy(self._line_cache)
            self._line_cache.clear()
            return res

        return 0


# 网络接口真是事情一大堆，碎片化令人烦躁
# noinspection PyBroadException
class _ThreadRPCWebAPIServer(threading.Thread):
    """
    线程化网络服务接口，使用多线程方式在不阻塞主线程情况下实现流畅的数据收发。
    指令的处理不是线程化的，不需要保持应用模块之间的并发安全性，除了通过队列读取和写入之外没有并发操作。
    API 地址参数:

        class:/path/to/api[,max_chunk_length,chunk_interval,transmit_timeout]

    # TODO 此服务方法无法支持多路同传模式，考虑使用转发工具或者其他方法
    """
    RESERVED_PREFIX = "api"

    def __init__(self, api_string: str):
        """
        初始化 API 服务器
        :param api_string: API 服务参数字符串
        """
        super().__init__()
        self.daemon = True

        self._topic_handler = LoggingTopicHandler(logging.INFO)
        self._tele = None

        self._api_path, self._rrt_kwargs = self._try_parse_path_str(api_string)

        self._commands = {}
        self._topics = {}
        self._active_topics = {}
        self._jsonrpc_in_pack = queue.Queue(maxsize=0)
        self._jsonrpc_out_pack = queue.Queue(maxsize=0)

        self._set_inline_methods()

    @staticmethod
    def _try_parse_path_str(_str: str) -> tuple[str, dict]:
        """
        解析带控制参数的服务器路径_BaseRemoteRequestTelemetry
        :param _str: 路径字符串
        :return: UNIX 套接字路径，块大小，块间延时，超时时间
        """
        if "," not in _str:
            return _str, {}

        if len(opts := _str.split(",")) != 4:
            raise MJRuntimeError(MJRuntimeError.INTERNAL_ERROR, "Invalid RPC server path string")

        else:
            try:
                return opts[0], {"max_chunk_length": int(opts[1]),
                                 "chunk_interval": float(opts[2]),
                                 "transmit_timeout": float(opts[3])}
            except ValueError:
                raise MJRuntimeError(MJRuntimeError.INTERNAL_ERROR, "Invalid RPC server path string")

    def _set_inline_methods(self):
        """内置指令和话题"""
        self._commands.setdefault("api/echo", self.command_ECHO)
        self._commands.setdefault("api/ls", self.command_LS)
        self._commands.setdefault("api/restart_mj", self.command_RESTART_SYSTEM)
        self._topics.setdefault("api/logging", self._topic_handler.topic_SYS_LOG)

    def _rpc_serve_forever(self):
        """读入数据包，尝试解析为 JSON 对象，将对象委托给请求处理，读取返回的 JSON 对象，打包发送。"""
        this_batch_cmds_ids = list()
        trans_wait_objs = list()

        # 服务端监听客户端调用，直到客户端请求调用，此函数初级筛选这些请求，产生一个必须回复列表，然后将所有 jsonrpc 放入处理队列
        # 队列在主进程中被处理，此函数收集 RPC 响应包，完成所有必须响应之后发送
        try:
            if self._api_path.startswith("socket:"):
                self._tele = UnixSocketRRT(self._api_path[7:], is_server=True, **self._rrt_kwargs)

            elif self._api_path.startswith("ipv4:"):
                self._tele = IPv4SocketRRT(self._api_path[5:], is_server=True, **self._rrt_kwargs)

            elif self._api_path.startswith("serial:"):
                self._tele = SerialRRT(self._api_path[7:], **self._rrt_kwargs)

        except ConnectionError as err:
            logger.error(f"failed to create connection, while {str(err)}")
            return

        try:

            while True:
                recv_rpc_obj = self._tele.receive_response_obj()

                this_batch_cmds_ids.clear()
                trans_wait_objs.clear()

                if bool(recv_rpc_obj):

                    if type(recv_rpc_obj) != list:
                        self._jsonrpc_in_pack.put(recv_rpc_obj)
                        if "id" in recv_rpc_obj:
                            this_batch_cmds_ids.append(recv_rpc_obj["id"])
                    else:
                        for single_pack in recv_rpc_obj:
                            self._jsonrpc_in_pack.put(single_pack)
                            if "id" in single_pack:
                                this_batch_cmds_ids.append(single_pack["id"])

                    while bool(this_batch_cmds_ids):
                        res_obj = self._jsonrpc_out_pack.get(block=True)

                        if bool(res_obj):
                            if res_obj["id"] in this_batch_cmds_ids:
                                this_batch_cmds_ids.remove(res_obj["id"])
                            trans_wait_objs.append(res_obj)

                self._tele.send_request_obj(trans_wait_objs)

        except (BrokenPipeError, OSError, ConnectionError, JSONDecodeError) as err:
            logger.warning(f"transmission error due to {str(err)} session disconnected")

        finally:
            self._tele.close()
            self._tele = None
            for k, v in self._active_topics.items():
                k("unsubscribe")
            self._active_topics.clear()

    def run(self):
        while True:
            self._rpc_serve_forever()

    ######################
    # API 内置指令和话题
    ######################
    @staticmethod
    def command_ECHO(rpc_params):
        return rpc_params

    @staticmethod
    def command_RESTART_SYSTEM(rpc_params):
        """
        产生一个退出系统异常，
        """
        raise MJRuntimeError(MJRuntimeError.EXIT_ERROR,
                             "Exit system by API server, once program exit, it will restart once.")

    def command_LS(self, rpc_params):
        return list(self._commands.keys()) + list(self._topics.keys())

    ###################
    # 外部控制接口
    ###################
    def handle_requests(self, blocking=False, timeout=None):
        """
        处理传入的 RPC 请求，此函数响应所有的指令，获取所有订阅转态并且判断是否发送订阅消息。
        这是一个为 MJ 调度器设计的专用接口，在程序大循环中由 MJ 调用。
        :param timeout: 超时
        :param blocking: 阻塞处理，启用则会等待待客户端断开连接或者客户端主动跳出循环，
        禁用则只请求队列中已有的内容
        """
        if (not bool(self._tele)) or (self._jsonrpc_in_pack.empty() and (not blocking)):
            return

        while bool(self._tele):
            # 这里异常处理比较多，但是还是很好理解的，记住，这个函数在主线程事件循环执行，唯一的通信就是队列收发
            # 除了收发不要放入不合法的 json，这里出什么事都和 API 服务器没关系
            try:
                jsonrpc_obj: dict = self._jsonrpc_in_pack.get(block=blocking, timeout=timeout)

            except queue.Empty:
                break

            else:
                logger.debug(f"handle json obj {jsonrpc_obj}")
                # command 和 message 处理
                if jsonrpc_obj["method"] in self._commands:
                    if "id" in jsonrpc_obj:
                        try:
                            mtd = self._commands.get(jsonrpc_obj["method"])
                            jsonrpc_res = deepcopy(MJ_RPC_COMMON_RESULT)
                            jsonrpc_res["id"] = jsonrpc_obj["id"]
                            jsonrpc_res["result"] = mtd(jsonrpc_obj["params"])
                            self._jsonrpc_out_pack.put(jsonrpc_res)
                        except Exception as err:
                            jsonrpc_res = deepcopy(MJ_RPC_ERROR_RESULT)
                            jsonrpc_res["error"] = {"code": MJ_RPC_ERROR_INTERNAL_ERROR_CODE,
                                                    "message": str(err)}
                            jsonrpc_res["id"] = jsonrpc_obj["id"]
                            self._jsonrpc_out_pack.put(jsonrpc_res)
                            raise err
                    else:
                        mtd = self._commands.get(jsonrpc_obj["method"])
                        mtd(jsonrpc_obj["params"])

                # 新 topic 订阅处理，订阅机制比较神奇，类似激活订阅主题，然后就能隔一段时间收到更新
                elif ("id" in jsonrpc_obj) and jsonrpc_obj["method"] in self._topics:

                    if ((jsonrpc_obj["params"] == "subscribe") and
                            ((t := self._topics.get(jsonrpc_obj["method"])) not in self._active_topics)):
                        try:
                            jsonrpc_res = deepcopy(MJ_RPC_COMMON_RESULT)
                            jsonrpc_res["result"] = t("subscribe")
                            jsonrpc_res["id"] = jsonrpc_obj["id"]
                            self._jsonrpc_out_pack.put(jsonrpc_res)
                        except Exception as err:
                            jsonrpc_res = deepcopy(MJ_RPC_ERROR_RESULT)
                            jsonrpc_res["error"] = {"code": MJ_RPC_ERROR_INTERNAL_ERROR_CODE,
                                                    "message": str(err)}
                            jsonrpc_res["id"] = jsonrpc_obj["id"]
                            self._jsonrpc_out_pack.put(jsonrpc_res)
                            raise err
                        else:
                            self._active_topics.setdefault(t, jsonrpc_obj["id"])

                    elif ((jsonrpc_obj["params"] == "unsubscribe") and
                          ((t := self._topics.get(jsonrpc_obj["method"])) in self._active_topics)):
                        try:
                            jsonrpc_res = deepcopy(MJ_RPC_COMMON_RESULT)
                            jsonrpc_res["result"] = t("unsubscribe")
                            jsonrpc_res["id"] = jsonrpc_obj["id"]
                            self._jsonrpc_out_pack.put(jsonrpc_res)
                        except Exception as err:
                            jsonrpc_res = deepcopy(MJ_RPC_ERROR_RESULT)
                            jsonrpc_res["error"] = {"code": MJ_RPC_ERROR_INTERNAL_ERROR_CODE,
                                                    "message": str(err)}
                            jsonrpc_res["id"] = jsonrpc_obj["id"]
                            self._jsonrpc_out_pack.put(jsonrpc_res)
                            raise err
                        else:
                            self._active_topics.pop(t)

                    else:
                        jsonrpc_res = deepcopy(MJ_RPC_ERROR_RESULT)
                        jsonrpc_res["error"] = {"code": MJ_RPC_ERROR_INVALID_PARAMS_CODE,
                                                "message": f"{jsonrpc_obj['method']} not valid topic operation."}
                        jsonrpc_res["id"] = jsonrpc_obj["id"]
                        self._jsonrpc_out_pack.put(jsonrpc_res)

                # 如果客户端访问邮箱，检查 topic 更新，最后发送邮箱检查回应，就能利用发送机制送出所有包
                elif ("id" in jsonrpc_obj) and jsonrpc_obj["method"] == "api/mailbox":

                    for t, tid in self._active_topics.items():
                        try:
                            jsonrpc_res = deepcopy(MJ_RPC_COMMON_RESULT)
                            jsonrpc_res["result"] = t("mailbox_update")
                            jsonrpc_res["id"] = tid
                            self._jsonrpc_out_pack.put(jsonrpc_res)
                        except Exception as err:
                            jsonrpc_res = deepcopy(MJ_RPC_ERROR_RESULT)
                            jsonrpc_res["error"] = {"code": MJ_RPC_ERROR_INTERNAL_ERROR_CODE,
                                                    "message": str(err)}
                            jsonrpc_res["id"] = jsonrpc_obj["id"]
                            self._jsonrpc_out_pack.put(jsonrpc_res)
                            raise err

                    jsonrpc_res = deepcopy(MJ_RPC_TOPIC_UPDATE_RESULT)
                    jsonrpc_res["result"] = f"{time.time()}"
                    jsonrpc_res["id"] = jsonrpc_obj["id"]
                    self._jsonrpc_out_pack.put(jsonrpc_res)

                # 只有 ID 没有方法的异常请求
                elif "id" in jsonrpc_obj:
                    jsonrpc_res = deepcopy(MJ_RPC_ERROR_RESULT)
                    jsonrpc_res["error"] = {"code": MJ_RPC_ERROR_MTD_NOT_FOUND_CODE,
                                            "message": f"{jsonrpc_obj['method']} not available"}
                    jsonrpc_res["id"] = jsonrpc_obj["id"]
                    self._jsonrpc_out_pack.put(jsonrpc_res)

    def disconnect(self):
        """
        清空对象引用，断开遥测连接
        """
        self._commands.clear()
        self._topics.clear()
        self._active_topics.clear()
        self._tele.close()

    def add_command(self, label: str, command_handler):
        return self.add_commands(**{label: command_handler})

    def add_commands(self, **kwargs: dict):
        for key, part in kwargs.items():
            if key.startswith(self.RESERVED_PREFIX):
                raise MJRuntimeError(MJRuntimeError.PROCESSOR_ERROR, f"{key} conflicts with system key.")
            if key in self._commands.keys():
                raise MJRuntimeError(MJRuntimeError.PROCESSOR_ERROR, f"{key} conflicts with exist key.")

            self._commands.setdefault(key, part)

    def add_topic(self, label: str, command_handler):
        return self.add_topics(**{label: command_handler})

    def add_topics(self, **kwargs: dict):
        for key, part in kwargs.items():
            if key.startswith(self.RESERVED_PREFIX):
                raise MJRuntimeError(MJRuntimeError.PROCESSOR_ERROR, f"{key} conflicts with system key.")
            if key in self._topics.keys():
                raise MJRuntimeError(MJRuntimeError.PROCESSOR_ERROR, f"{key} conflicts with exist key.")

            self._topics.setdefault(key, part)


MJWebApiServer = _ThreadRPCWebAPIServer
